package org.apache.flink.kubernetes.kubeclient.factory;

import static org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.KUBERNETES_KERBEROS_MOUNT_DECORATOR_ENABLED;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.kubeclient.ComponentPod;
import org.apache.flink.kubernetes.kubeclient.KubernetesConsumerManagerSpecification;
import org.apache.flink.kubernetes.kubeclient.decorators.CmdConsumerManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.ERConfMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.EnvSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.InitConsumerManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.KerberosMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.KubernetesStepDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.MountSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.PodTemplateMountDecorator;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesMasterComponentParameters;
import org.apache.flink.kubernetes.kubeclient.resources.KubernetesOwnerReference;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
import org.apache.flink.util.Preconditions;

import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;

/**
 * @类名: KubernetesResourceConsumerFactory
 * @说明: k8s资源消费者工厂，构建资源消费者规格，包括Deployment对象
 *
 * @author   leehom
 * @Date	 2024年1月18日 上午11:41:33
 * 修改记录：
 *
 * @see 	 
 */
/**
 * Utility class for constructing all the Kubernetes components on the client-side. This can include
 * the Deployment, the ConfigMap(s), and the Service(s).
 */
public class KubernetesResourceConsumerFactory {

	// 构建资源管理器部署规格, 包括Deployment，ConfigMap，Service
    public static KubernetesConsumerManagerSpecification buildKubernetesResourceConsumerSpecification(
            ComponentPod podTemplate, KubernetesMasterComponentParameters kubernetesMasterComponentParameters)
            throws IOException {
    	// 复制pod模板，后面自定制
        ComponentPod componentPod = Preconditions.checkNotNull(podTemplate).copy();
        // 搭载资源
        List<HasMetadata> accompanyingResources = new ArrayList<>();
        // 构建装饰器链，按顺序
        final List<KubernetesStepDecorator> stepDecorators =
                new ArrayList<>(
                        Arrays.asList(
                                new InitConsumerManagerDecorator(kubernetesMasterComponentParameters),
                                new EnvSecretsDecorator(kubernetesMasterComponentParameters),
                                new MountSecretsDecorator(kubernetesMasterComponentParameters),
                                new CmdConsumerManagerDecorator(kubernetesMasterComponentParameters),
                                new InternalServiceDecorator(kubernetesMasterComponentParameters),
                                new ExternalServiceDecorator(kubernetesMasterComponentParameters))
                        );
        //
        Configuration configuration = kubernetesMasterComponentParameters.getConfiguration();
        if (configuration.getBoolean(KUBERNETES_KERBEROS_MOUNT_DECORATOR_ENABLED)) {
            stepDecorators.add(new KerberosMountDecorator(kubernetesMasterComponentParameters));
        }
        //
        stepDecorators.addAll(
                Arrays.asList(
                        // new ERConfMountDecorator(kubernetesMasterComponentParameters),
                        new PodTemplateMountDecorator(kubernetesMasterComponentParameters)));
        // 责任链模式，增加和修改flink pod特性，其他搭载资源
        for (KubernetesStepDecorator stepDecorator : stepDecorators) {
            componentPod = stepDecorator.decoratePod(componentPod);
            accompanyingResources.addAll(stepDecorator.buildAccompanyingKubernetesResources());
        }
        // 构建k8s部署对象
        final Deployment deployment =
                createConsumerManagerDeployment(componentPod, kubernetesMasterComponentParameters);
        // 合成规格
        return new KubernetesConsumerManagerSpecification(deployment, accompanyingResources);
    }

    // 创建资源消费者k8s Deployment
    private static Deployment createConsumerManagerDeployment(
            ComponentPod componentPod, KubernetesMasterComponentParameters kubernetesJobManagerParameters) {
        final Container resolvedMainContainer = componentPod.getMainContainer();

        final Pod resolvedPod =
                new PodBuilder(componentPod.getPodWithoutMainContainer())
                        .editOrNewSpec()
                        .addToContainers(resolvedMainContainer)
                        .endSpec()
                        .build();

        return new DeploymentBuilder()
                .withApiVersion(Constants.APPS_API_VERSION)
                .editOrNewMetadata()
                .withName(
                        KubernetesUtils.getDeploymentName(
                                kubernetesJobManagerParameters.getClusterId()))
                .withAnnotations(kubernetesJobManagerParameters.getAnnotations())
                .withLabels(kubernetesJobManagerParameters.getLabels())
                .withOwnerReferences(
                        kubernetesJobManagerParameters.getOwnerReference().stream()
                                .map(e -> KubernetesOwnerReference.fromMap(e).getInternalResource())
                                .collect(Collectors.toList()))
                .endMetadata()
                .editOrNewSpec()
                .withReplicas(kubernetesJobManagerParameters.getReplicas())
                .editOrNewTemplate()
                .withMetadata(resolvedPod.getMetadata())
                .withSpec(resolvedPod.getSpec())
                .endTemplate()
                .editOrNewSelector()
                .addToMatchLabels(kubernetesJobManagerParameters.getSelectors())
                .endSelector()
                .endSpec()
                .build();
    }
}
